package com.atguigu.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Felix
 * @date 2023/12/13
 * 该案例演示了FlinkSQL创建表 执行查询 以及输出
 */
public class FlinkSQL01_Base {
    public static void main(String[] args) throws Exception {
        // 创建流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance()
        //     .inStreamingMode()
        //     .build();
        // StreamTableEnvironment.create(env, environmentSettings);

        // 创建动态表        流--->表
        tableEnv.executeSql("CREATE TABLE source ( \n" +
            "    id INT, \n" +
            "    ts BIGINT, \n" +
            "    vc INT \n" +
            ") WITH ( \n" +
            "    'connector' = 'datagen', \n" +
            "    'rows-per-second'='1', \n" +
            "    'fields.id.kind'='random', \n" +
            "    'fields.id.min'='1', \n" +
            "    'fields.id.max'='10', \n" +
            "    'fields.ts.kind'='sequence', \n" +
            "    'fields.ts.start'='1', \n" +
            "    'fields.ts.end'='1000000', \n" +
            "    'fields.vc.kind'='random', \n" +
            "    'fields.vc.min'='1', \n" +
            "    'fields.vc.max'='100'\n" +
            ")\n");
        //从表中获取数据       持续查询
        // TableResult tableResult = tableEnv.executeSql("select * from source");
        //打印到控制台
        // tableResult.print();
        Table sourceTable = tableEnv.sqlQuery("select * from source");
        sourceTable.execute().print();

        //注册表对象 到表执行环境中
        // tableEnv.createTemporaryView("source_table",sourceTable);
        // tableEnv.executeSql("select * from source_table").print();
        //直接查询表对象  会报错
        // tableEnv.executeSql("select * from sourceTable").print();
        // tableEnv.executeSql("select * from " + sourceTable).print();

        //通过TableAPI查询表中数据
        /*tableEnv
            .from("source")
            .where($("id").isEqual(1))
            .select($("id"),$("vc")).execute().print();*/


        tableEnv.executeSql("CREATE TABLE sink (\n" +
            "    id INT, \n" +
            "    ts BIGINT, \n" +
            "    vc INT\n" +
            ") WITH (\n" +
            "'connector' = 'print'\n" +
            ");");

        // SQL方式向表中插入数据
        // tableEnv.executeSql("insert into sink select * from source");

        // TableAPI方式向表中插入数据
        tableEnv.sqlQuery("select * from source").executeInsert("sink");

        tableEnv.sqlQuery("select id,vc from sink").execute().print();


        //注意：如果是API，必须通过execute提交job，但是最后操作的是动态表，那么不需要通过execute提交
        // env.execute();
    }
}
